package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.article.ArticleConstans;
import com.heima.model.article.vo.HotArticleVo;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Calendar;
import java.util.Comparator;
import java.util.Date;
import java.util.List;

@Component
public class HotArticleListener {

    @Autowired
    RedisTemplate redisTemplate;

    //@KafkaListener(topics = {"article-kafka"})
    public void listener(ConsumerRecord<?, ?> record) {
        System.out.println(record.key() + "===========" + record.value());

        //1.取出数据，只取前五天
        Object value = record.value();
       // String key = (String) record.key();
        HotArticleVo hotArticleVo = JSON.parseObject(value.toString(), HotArticleVo.class);
        if (hotArticleVo.getPublishTime().getTime() <= getDate().getTime()) {
            return;
        }
        Object o = redisTemplate.boundValueOps(ArticleConstans.HOT_ARTICLE_FIRST_PAGE + hotArticleVo.getChannelId()).get();

        List<HotArticleVo> hotArticleVos = JSON.parseArray(o.toString(), HotArticleVo.class);

        //2.判断redis每个频道中数据是否有三十条，如果有进行3.1，如果没有则直接添加
        if (hotArticleVos.size() < 30) {
            redisTemplate.boundValueOps(ArticleConstans.HOT_ARTICLE_FIRST_PAGE + hotArticleVo.getChannelId()).set(hotArticleVo);
        }

        //判断该集合是否包含该文章如果包含则删除，在添加
        if (hotArticleVos.contains(hotArticleVo)) {
            hotArticleVos.remove(hotArticleVo);
            hotArticleVos.add(hotArticleVo);
        } else {
            //3.1取出最小数据与redis中数据积分对比
            HotArticleVo hotArticleVo1 = hotArticleVos.get(hotArticleVos.size() - 1);
            if (hotArticleVo1.getScore() < hotArticleVo.getScore()) {
                //3.2如果现有数据积分大于redis中最小数据，则删除最小数据，保存该数据到redis中
                hotArticleVos.remove(hotArticleVos.size()-1);
                hotArticleVos.add(hotArticleVo);
            }
        }
        //添加后重新排序
        hotArticleVos.sort(new Comparator<HotArticleVo>() {
            @Override
            public int compare(HotArticleVo o1, HotArticleVo o2) {
                //降序
                return o2.getScore()-o1.getScore();
            }
        });

       redisTemplate.boundValueOps(ArticleConstans.HOT_ARTICLE_FIRST_PAGE+hotArticleVo.getChannelId()).set(hotArticleVos);

    }

    private Date getDate() {
        Calendar instance = Calendar.getInstance();
        instance.set(Calendar.DAY_OF_YEAR, instance.get(Calendar.DAY_OF_YEAR) - 5);
        instance.set(Calendar.HOUR_OF_DAY, 1);
        instance.set(Calendar.MINUTE, 0);
        instance.set(Calendar.SECOND, 0);
        return instance.getTime();
    }
}